package com.example.reactstream;

import java.util.concurrent.Flow;
import java.util.concurrent.SubmissionPublisher;


/**
 * publisher->subscriber 的关系可以参考/pic/publisher-processor-subscriber.png 这张图
 */
public class Reactstreamdemo01 {

    public static void main(String[] args) throws InterruptedException {

        //1，创建一个发布者
        SubmissionPublisher publisher =new SubmissionPublisher();

        //2，创建一个订阅者
        Flow.Subscriber subscriber=new Flow.Subscriber() {

            private Flow.Subscription subscription;

            @Override
            public void onSubscribe(Flow.Subscription subscription) {
                this.subscription=subscription;
                System.out.println("建立订阅关系");
                subscription.request(1);  //第一次需要
            }
            //订阅者接收发布者的数据都需要这个方法
            @Override
            public void onNext(Object o) {
                System.out.println("订阅者接收数据: " + o);
                // TODO: 业务方法
                subscription.request(10);  //订阅者处理完数据还需要接着处理多少条进行处理  这就是背压，也就是订阅者处理多少条数据不是由发布者控制的，而是由这个背压控制的

            }

            @Override
            public void onError(Throwable throwable) {
                System.out.println("发生异常时");
            }

            @Override
            public void onComplete() {
                System.out.println("数据处理完毕调用");

            }
        };

        //3，建立订阅者和发布者的关系，这里稍微多说一下，这里的订阅发起者是发布者，跟mq有写不同
        publisher.subscribe(subscriber);

        for (int i = 0; i < 20; i++) {
            publisher.submit("hello react stream "+ i);
        }

        Thread.currentThread().join();


    }
}
